package com.huahua.bigdata.sparksql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.spark.sql.*;
import org.apache.spark.sql.expressions.Aggregator;

import java.io.Serializable;

public class SparkSQL03_SQL_UDAF {
    public static void main(String[] args) {
        // TODO 构建环境对象
        //   Spark在结构化数据的处理场景中对核心功能, 环境进行了封装
        //   构建SparkSQL的环境对象时，一般采用构建器模式
        final SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("SparkSQL")
                .getOrCreate();

        final Dataset<Row> ds = sparkSession.read().json("data/user.json");
        ds.createOrReplaceTempView("user");

        // TODO SparkSQL采用特殊的方式将UDAF转换成UDF使用
        //        UDAF使用时需要创建自定义聚合对象
        //              第一个参数表示UDAF对象
        //              第二个参数表示
        sparkSession.udf().register("avgAge", functions.udaf(
                new MyAvgAgeUDAF(), Encoders.LONG()
        ));
        String sql = "select avgAge(age) from user";
        Dataset<Row> sqlDS = sparkSession.sql(sql);
        sqlDS.show();

        // TODO 释放资源
        sparkSession.close();
    }
}
